package com.seezoon.infrastructure.tcp.handler;

import com.seezoon.infrastructure.exception.Assertion;
import com.seezoon.infrastructure.tcp.codec.JTMessage;
import com.seezoon.infrastructure.tcp.codec.JTSerialization;
import com.seezoon.infrastructure.tcp.common.RpcContext;
import com.seezoon.infrastructure.tcp.session.Session;
import com.seezoon.infrastructure.tcp.session.SessionManager;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.slf4j.MDC;
import org.springframework.util.Assert;

/**
 * 也可以响应读写事件 ，使用ChannelDuplexHandler
 */
@Slf4j
@Sharable
public class NettyServerHandler extends SimpleChannelInboundHandler<JTMessage> {

    private final DispatcherHandler dispatcher;
    private final ExecutorService workerPool;

    public NettyServerHandler(DispatcherHandler dispatcher, ExecutorService workerPool) {
        Assert.notNull(dispatcher, "dispatcher must not be null");
        Assert.notNull(workerPool, "workerPool must not be null");
        this.dispatcher = dispatcher;
        this.workerPool = workerPool;
    }

    /**
     * 断开链接
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //连接关闭时候后
        Session session = getSession(ctx);
        if (null != session) {
            session.invalidate();
            log.debug("channelInactive clientNo:{},channel:{}", session.getClientNo(), ctx.channel());
        }
        ctx.close();
        log.debug("channelInactive channel:{}", ctx.channel());
    }

    public void doChannelRead(ChannelHandlerContext ctx, JTMessage msg) {
        // 创建session
        Session session = createSessionIfAbsent(msg.getSerialNo(), ctx);
        // 鉴权
        if (msg.needAuth() && !session.isAuthed()) {
            log.error("not authorized channel:{} jtmsg:{}", ctx.channel(), msg);
            ctx.close();
        }

        MessageHandler messageHandler = dispatcher.dispatch(msg.getMsgId());
        // 服务端读取客户端请求
        if (messageHandler != null) {
            long start = System.currentTimeMillis();
            Object request = JTSerialization.decode(msg.getBody(), messageHandler.getRequestType());
            RpcContext rpcContext = creatRpcContext(msg, session.getClientId());
            Object response = messageHandler.execute(request, rpcContext);

            ctx.writeAndFlush(
                    createWriteMessage(messageHandler.responseType(), session.nextSequence(), rpcContext,
                            response));
            if (log.isInfoEnabled()) {
                log.info("server channelRead request:{}, channelWrite response:{},use {} ms channel:{}", request,
                        response, System.currentTimeMillis() - start, ctx.channel());
            }


        } else { // 客户端响应
            // 响应消息2个字节为消息流水号
            if (msg.getBody().readableBytes() < 2) {
                log.error("server channelRead responseMsgSerialNo error lenght < 2 ,jtmessage:{},channel:{}", msg,
                        ctx.channel().id());
                return;
            }
            int responseMsgSerialNo = msg.getBody().getUnsignedShort(0);
            ResponseFuture responseFuture = FutureManager.getServerFutureManager()
                    .remove(ctx.channel().id().asLongText(),
                            responseMsgSerialNo);
            if (responseFuture == null) { // 没有找到消息
                log.error("server channelRead no mapping responseMsgSerialNo:{},jtmessage:{},channel:{}",
                        responseMsgSerialNo,
                        msg, ctx.channel());
                return;
            }
            Object response = JTSerialization.decode(msg.getBody(), responseFuture.getClazz());
            if (log.isDebugEnabled()) {
                log.debug("server channelRead response:{} jtmessage:{},channel:{}", response, msg, ctx.channel());
            }
            responseFuture.getFuture().complete(response);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, JTMessage msg) throws Exception {
        try {
            workerPool.execute(() -> {
                try {
                    MDC.put("tid", RandomStringUtils.randomAlphanumeric(10));
                    this.doChannelRead(ctx, msg);
                    MDC.remove("tid");
                } catch (Exception e) {
                    log.error("server channelRead exception:{}", msg, e);
                    ctx.close();
                }
            });
        } catch (RejectedExecutionException e) {
            log.error("server channelRead reject msg:{},channel:{}", msg, ctx.channel());
            throw e;
        }
    }

    /**
     * 握手完成
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        log.debug("channelActive channel:{}", ctx.channel());
    }

    /**
     * 响应netty 心跳  这个可以给客户端发送一些特定包 来标识
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            Session session = getSession(ctx);
            if (null != session) {
                session.invalidate();
                log.debug("userEventTriggered clientNo:{} channel:{}", session.getClientNo(), ctx.channel());
            }
            log.debug("userEventTriggered channel:{}", ctx.channel());
            ctx.close();
            return;
        }
        super.userEventTriggered(ctx, evt);
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("exceptionCaught channel:{}", ctx.channel(), cause);
        ctx.close();
    }


    private Session getSession(ChannelHandlerContext ctx) {
        Session session = ctx.channel().attr(Session.key).get();
        return session;
    }

    private Session createSessionIfAbsent(String serialNo, ChannelHandlerContext ctx) {
        Assertion.notEmpty(serialNo, "serialNo must not be empty");
        Assertion.notNull(ctx, "ctx must not be null");
        Session session = ctx.channel().attr(Session.key).get();
        if (null == session) {
            session = new Session(serialNo, ctx.channel());
            SessionManager.getInstance().add(serialNo, session);
        }
        return session;
    }


    private RpcContext creatRpcContext(JTMessage msg, Long clientId) {
        RpcContext rpcContext = new RpcContext();
        rpcContext.setClientId(clientId);
        rpcContext.setMsgId(msg.getMsgId());
        rpcContext.setSerialNo(msg.getSerialNo());
        rpcContext.setMsgSerialNo(msg.getMsgSerialNo());
        rpcContext.setDataLength(msg.getBodyLength());
        rpcContext.setPackageTotal(msg.getPackageTotal());
        rpcContext.setPackageNo(msg.getPackageNo());
        return rpcContext;
    }

    /**
     * @param msgId 回包消息ID
     * @param msgSerialNo 回包消息序号
     * @param context
     * @param response
     * @return
     */
    private JTMessage createWriteMessage(int msgId, int msgSerialNo, RpcContext context, Object response) {
        byte[] encoded = JTSerialization.encode(response);
        // 服务单回包不考虑分包和加密
        JTMessage out = new JTMessage();
        out.setMsgId(msgId);
        out.setMsgProperties(encoded.length);
        out.setSerialNo(context.getSerialNo());
        out.setMsgSerialNo(msgSerialNo);
        out.setBody(Unpooled.wrappedBuffer(encoded));
        return out;
    }

}
